package dper.study.codes.hbase.service.impl;

import com.google.common.base.Strings;
import dper.study.codes.elasticsearch.dao.mapper.GitRepoMapper;
import dper.study.codes.elasticsearch.model.GitRepoModel;
import dper.study.codes.hbase.api.HGitRepoQueryService;
import dper.study.codes.hbase.consts.GitRepoConsts;
import dper.study.codes.hbase.mapper.HGitRepoRowMapper;
import dper.study.codes.hbase.util.RowKeyUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * Created by leichen on 2017/4/18.
 */
@Service("hGitRepoQueryServiceImpl")
public class HGitRepoQueryServiceImpl implements HGitRepoQueryService {

    private Logger logger = Logger.getLogger(HGitRepoQueryServiceImpl.class) ;

    @Autowired
    private HbaseTemplate hbaseTemplate;

    public GitRepoModel queryByGId(int gitId) {
        String rowKey =  RowKeyUtil.buildGRRowKey(gitId);
        if (StringUtils.isBlank(rowKey))
            return null;
        return hbaseTemplate.get(GitRepoConsts.TABLE_NAME, rowKey,
                new HGitRepoRowMapper());
    }

    public List<GitRepoModel> queryByGIdList(final List<Integer> gitIdList) {
        List<GitRepoModel> ret = null;
        hbaseTemplate.execute(GitRepoConsts.TABLE_NAME, new TableCallback<List<GitRepoModel>>() {
            public List<GitRepoModel> doInTable(HTableInterface hTableInterface) throws Throwable {
                List<Get> getList = new ArrayList<Get>();
                for(Integer gitId:gitIdList){
                    String rowKey =  RowKeyUtil.buildGRRowKey(gitId);
                    Get get = new Get(Bytes.toBytes(rowKey)) ;
                    get.addColumn(Bytes.toBytes(GitRepoConsts.FAMILY_GITREPO_INFO),
                            Bytes.toBytes(GitRepoConsts.COLUMN_GITREPO_CONTENT)) ;
                    getList.add(get) ;
                }
                List<GitRepoModel> ret =new ArrayList<GitRepoModel>();
                Result[] result = hTableInterface.get(getList);
                for (int i = 0; i < result.length; i++) {
                    HGitRepoRowMapper mapper = new HGitRepoRowMapper();
                    GitRepoModel gitRepoModel = mapper.mapRow(
                            result[i], i);
                    ret.add(gitRepoModel);
                }
                return ret ;
            }
        }) ;
        return null;
    }

    public List<GitRepoModel> queryAll(int authorId) {
        long endTime = new Date().getTime();
        long startTime = endTime - 1000 * 60 * 60 * 24 * 2;
        return queryAllByUpdateTimeRange(authorId,startTime,endTime) ;
    }

    public List<GitRepoModel> queryAllByUpdateTimeRange(int authorId, long startTime, long endTime) {
        List<GitRepoModel> ret = null ;
        String startRowKey = RowKeyUtil.buildGRRowKey(authorId, new Date(
                endTime));
        String stopRowKey = RowKeyUtil.buildGRRowKey(authorId, new Date(
                startTime));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startRowKey));
        scan.setStopRow(Bytes.toBytes(stopRowKey));
        scan.setMaxResultSize(3000);
        ret = hbaseTemplate.find(GitRepoConsts.TABLE_NAME,
                scan, new HGitRepoRowMapper());
//        if(ret != null && !ret.isEmpty()){
//            Collections.sort(ret, new Comparator<GitRepoModel>() {
//                public int compare(GitRepoModel o1, GitRepoModel o2) {
//                    return o1.getUpdateTime().getTime()-o2.getUpdateTime().getTime() > 0 ? 1 :-1;
//                }
//            });
//        }
        return ret ;
    }

    public List<GitRepoModel> queryAllByUpdateTimeRangeUsingRedis(int authorId, long startTime, long endTime) {
        return null;
    }

    public List<GitRepoModel> queryRowPrefix(int authorId, int querySize) {
        List<GitRepoModel> ret = null ;
        String rowKeyPrefix = RowKeyUtil.buildGRRowKey(authorId,"");
        Scan scan = new Scan();
        scan.setRowPrefixFilter(Bytes.toBytes(rowKeyPrefix));
        scan.setMaxResultSize(querySize);
        ret = hbaseTemplate.find(GitRepoConsts.TABLE_NAME,
                scan, new HGitRepoRowMapper());
        return ret ;
    }
}
